package com.atguigu.sparkstreaming.output

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import redis.clients.jedis.Jedis

/**
 * Created by Smexy on 2022/7/15
 *
 *    演示把wordcount结果存入redis
 *
 *        用string去存:
 *                      一个单词是一个key
 *        用hash去存:
 *                   key: wordcount
 *                   value: hash
 *                            [filed,value]
 *                             word , count
 *
 *   -------------------------
 *   def foreachRDD(foreachFunc: (RDD[T]) => Unit): Unit
 *      ds2.foreachRDD() 和transform是类似的，都是将对DStream的运算转为对RDD的运算。
 *      foreachFunc： 没有返回值的函数。
 *
 *      foreachRDD： 没有返回值。
 *
 *   -------------------------------
 *
 *        以partition为单位操作(创建Connection)！
 *
 *      读数据库:   RDD.mapPartition
 *                  有返回值
 *
 *      写数据库:   RDD.foreachPartition
 *                  没返回值
 *
 *
 */
object ForeachRDDDemo {

  def main(args: Array[String]): Unit = {

    val streamingContext = new StreamingContext("local[*]", "ForeachRDDDemo", Seconds(5))

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "220309",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "true"
    )


    val topics = Array("topicA")

    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )



    //(String, Int):(word,n)
    val ds2: DStream[(String, Int)] = ds.flatMap(record => record.value().split(" "))
      .map((_, 1)).reduceByKey(_+_)

    ds2.cache()

    ds2.print(1000)
    ds2.foreachRDD(rdd => rdd.foreachPartition(partition => {

      //一个分区创建一个连接
      val jedis = new Jedis("hadoop102", 6379)

      //使用连接
      partition.foreach{
        case (word, count) => {

          val state: String = jedis.hget("wordcount", word)

          if (state == null){
            jedis.hset("wordcount", word,count.toString)
          }else{
            jedis.hset("wordcount", word,(count + state.toInt).toString)
          }

        }
      }

      //关闭连接
      jedis.close()
    }))

    // 启动APP
    streamingContext.start()

    // 阻塞进程，让进程一直运行
    streamingContext.awaitTermination()

  }

}
